package com.niodata.dp.log.viewer;

import com.google.common.base.Strings;
import com.jcraft.jsch.ChannelExec;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.Session;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.log4j.Logger;

public class JschViewerSession implements ViewerSession {

  private static final Logger logger = Logger.getLogger(JschViewerSession.class);

  private Thread dataReceiveThread;
  private Thread dataOutputThread;
  private Session session;
  private ChannelExec channel;
  private volatile boolean closed = false;
  private final Host host;
  private final String cmd;
  private PrintWriter streamWriter;
  private volatile boolean streamCompleted = false;
  private ConcurrentLinkedQueue<String> dataQueue;


  public JschViewerSession(Host host, String cmd) {
    this.host = host;
    this.cmd = cmd;
  }

  /**
   * open session.
   *
   * @throws IOException e
   */
  public void open() throws IOException {
    if (closed) {
      throw new RuntimeException("session has been closed");
    }
    JSch jsch = new JSch();
    try {
      if (!Strings.isNullOrEmpty(host.getIdrsaPath())) {
        jsch.addIdentity(host.getIdrsaPath());
      }
      session = jsch.getSession(host.getUser(), host.getIp(), host.getPort());
      session.setConfig("PreferredAuthentications", "publickey,password");
      if (!Strings.isNullOrEmpty(host.getPassword())) {
        session.setPassword(host.getPassword());
      }
      session.setConfig("StrictHostKeyChecking", "no");
      session.connect();
      channel = (ChannelExec) session.openChannel("exec");
      channel.setPty(true);
      InputStream in = channel.getInputStream();
      channel.setCommand(cmd);
      logger.info("host:" + host.getIp() + "," + host.getUser() + ",cmd:" + cmd);
      channel.connect();
      dataQueue = new ConcurrentLinkedQueue<>();
      dataReceiveThread = new Thread() {
        @Override
        public void run() {
          BufferedReader reader = null;
          try {
            int count = 0;
            reader = new BufferedReader(new InputStreamReader(in));
            String line = null;
            while (null != (line = reader.readLine())) {
              dataQueue.add(line);
              count++;
              if (count == 100) {
                sleep(10);
                count = 0;
              }
              if (closed) {
                break;
              }
            }
            streamCompleted = true;
          } catch (Exception e) {
            return;
          } finally {
            try {
              close();
              reader.close();
            } catch (IOException e) {
              //e.printStackTrace();
            }
          }
        }
      };
      dataReceiveThread.start();
    } catch (Exception e) {
      logger.error("open session error", e);
      throw new IOException(e);
    }
  }

  @Override
  public void close() throws IOException {
    closed = true;
    dataReceiveThread.interrupt();
    channel.disconnect();
    session.disconnect();
  }

  @Override
  public void clear() {
    if (null != dataQueue) {
      dataQueue.clear();
    }
  }

  @Override
  public boolean isClosed() {
    return closed;
  }

  @Override
  public Host getHost() {
    return host;
  }

  @Override
  public List<String> readLines(int count) {
    List<String> lines = new ArrayList<>(count);
    for (int i = 0; i < count; i++) {
      if (!dataQueue.isEmpty()) {
        lines.add(dataQueue.remove());
      } else {
        break;
      }
    }
    return lines;
  }

  @Override
  public List<String> readLines(int count, int waitTime) {
    if (this.dataQueue.isEmpty()) {
      waitForQueueData(waitTime);
    }
    return this.readLines(count);
  }

  @Override
  public String readLine() {
    if (!dataQueue.isEmpty()) {
      return dataQueue.remove();
    }
    return null;
  }

  @Override
  public String readLine(int waitTime) {
    if (dataQueue.isEmpty()) {
      waitForQueueData(waitTime);
    }
    return this.readLine();
  }

  private void waitForQueueData(int waitTime) {
    int times = waitTime / 10;
    while (times > 0) {
      if (!dataQueue.isEmpty()) {
        return;
      } else {
        try {
          Thread.sleep(10);
        } catch (InterruptedException e) {
          return;
        }
        times--;
      }
    }
  }

  @Override
  public boolean hasMore() {
    return !dataQueue.isEmpty() || !streamCompleted;
  }

  @Override
  public synchronized void bindStreamToOutput(PrintWriter printWriter) {
    if (this.streamWriter != null) {
      throw new RuntimeException("stream already binded");
    }
    this.streamWriter = printWriter;
    this.dataOutputThread = new DataOutputThread();
    dataOutputThread.start();
  }

  private class DataOutputThread extends Thread {

    public void run() {
      while (!streamCompleted || dataQueue.isEmpty()) {
        {
          try {
            if (dataQueue.isEmpty()) {
              sleep(200);
              continue;
            }
            streamWriter.write(dataQueue.remove() + "\r\n");
          } catch (Exception e) {
            break;
          }
        }
      }
    }
  }
}
